package com.aotain.coeus.spark;

import java.net.URI;
import java.util.ArrayList;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.log4j.Logger;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.api.java.function.PairFunction;

import com.aotain.common.CommonFunction;

import scala.Tuple2;

/**
 * 
 * DNS数据处理
 * <功能详细描述>
 * 处理DPI UDII中nonehttp输出的DNS数据，将host与IP关联关系写入HDFS中
 * 
 * @author  Turk
 * @version  [v1.0, 2017年2月9日]
 * @see  [相关类/方法]
 * @since  [产品/模块版本]
 */
public class DNSDataProcess {

	public static void main(String[] args) {
		   //System.out.println("0###OK################################");
		
		if (args.length != 2){
		   System.err.printf("Usage: <SourcePath> <OutputPath>");
		   System.exit(1);
		}          
		
		//HTTPDCSpark spark = new HTTPDCSpark();
		int nexit = SparkFunc(args);
		   //System.out.println("1###OK################################");
		System.exit(nexit);
	}
	
	private static int SparkFunc(String[] args) {
		
		try 
		{
			SparkConf sparkConf = new SparkConf().setAppName("DNSDataProcess");
			Configuration config = new Configuration();
			JavaSparkContext ctx = new JavaSparkContext(sparkConf);
			 
			String sourcePath = args[0];
			String outputPath = args[1];
			 
			System.out.println("sourcePath: " + sourcePath);
			 
			
			//目标目录维护
			//String targetPath = args[1];
			System.out.println("targetPath: " + outputPath);
			FileSystem fsTarget = FileSystem.get(URI.create(outputPath),config);
			Path pathTarget = new Path(outputPath);
			if(fsTarget.exists(pathTarget))
			{
				fsTarget.delete(pathTarget, true);
				System.out.println("Delete path " + outputPath);
			}
			
			 JavaPairRDD<LongWritable,Text> tt = ctx.newAPIHadoopFile(sourcePath,TextInputFormat.class,
					 LongWritable.class, Text.class,config);
			 
			 //JavaPairRDD<String, Integer> s = tt.flatMap();
			 
			 JavaPairRDD<String, Integer> s = tt.flatMapToPair(new PairFlatMapFunction<Tuple2<LongWritable, Text>, String, Integer>() {

						/**
						 * serialVersionUID
						 */
						private static final long serialVersionUID = 1L;

						@Override
						public Iterable<Tuple2<String, Integer>> call(
								Tuple2<LongWritable, Text> arg0)
								throws Exception {
							// TODO Auto-generated method stub
							
							String[] items = arg0._2.toString().split("\\|",-1);
							
							ArrayList<Tuple2<String, Integer>> ret = new ArrayList<Tuple2<String, Integer>>();
							
							if(items.length > 28 && items[10].equals("53"))
							{
								String domain = items[27];
								//处理域名，只保留顶级域名
								String[] strArr = domain.split("\\.",-1);
								if(strArr.length >= 2)
								{//根域名//顶级域名									 
									 if(CommonFunction.IsSpecialRoot(domain) && strArr.length >= 3)
									 {//特殊处理例如 .com.cn  .com.hk
										 domain = String.format("%s.%s.%s", strArr[strArr.length-3],
												 strArr[strArr.length-2],strArr[strArr.length-1]);
									 }
									 else if(CommonFunction.IsSpecialRoot2Level(domain))
									 {
										 domain = String.format("%s.%s", 
												 strArr[strArr.length-2],strArr[strArr.length-1]);
									 }
									 else
									 {
										 return ret;
									 }
								}
								
								
								int nIPNum = 0;
								try
								{
									nIPNum = Integer.parseInt(items[28]);
								} catch (Exception ex) {
									Logger.getLogger(DNSDataProcess.class).error(arg0._2,ex);
									return ret;
								}
								
								
								int nFirstPos = 28;
								for(int i = 0; i<nIPNum; i++){
									String ip = items[nFirstPos + i*2 + 2];
									try
									{
										ip = CommonFunction.int2ip(Long.parseLong(ip));
									} catch (Exception ex) {
										Logger.getLogger(DNSDataProcess.class).error(arg0._2,ex);
										return ret;
									}
									String row = String.format("%s,%s", domain, ip);
									Tuple2<String, Integer> tuple = new Tuple2<String, Integer>(row, 1);
									ret.add(tuple);
								}
								return ret;
							}
							return ret;
						}				 
			 });
			
			 JavaPairRDD<String, Integer> clac = s.filter(new Function<Tuple2<String, Integer>, Boolean>(){

					/**
				 * serialVersionUID
				 */
				private static final long serialVersionUID = 1L;

					@Override
					public Boolean call(Tuple2<String, Integer> arg0) throws Exception {
						// TODO Auto-generated method stub
						if(arg0 == null)
							return false;
						else
							return true;
					}}).reduceByKey(new Function2<Integer, Integer, Integer>(){

						@Override
						public Integer call(Integer v1, Integer v2)
								throws Exception {
							// TODO Auto-generated method stub
							return v1 + v2;
						}
						
					});
			JavaPairRDD<String, String> result = clac.mapToPair(new PairFunction<Tuple2<String, Integer>, String, String>(){

				@Override
				public Tuple2<String, String> call(Tuple2<String, Integer> arg0)
						throws Exception {
					// TODO Auto-generated method stub
					String row = String.format("%s,%d,", arg0._1, arg0._2);
					return new Tuple2<String, String>(row,"");
				}
				
			});
			 
			result.saveAsHadoopFile(outputPath, Text.class, Text.class, TextOutputFormat.class);
				    	
			ctx.stop();
			
			return 0;
		} catch (Exception ex) {
			ex.printStackTrace();
			return 1;
		}
		
		
		
	}
}
